package com.stu.thread.demo.producer_consumer_model;

import com.stu.thread.demo.message_queue.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.ObjectUtils;

import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;

/**
 * @projectName: learn_used_code
 * @packageName: com.stu.thread.demo.producer_consumer_model
 * @className: com.stu.thread.demo.producer_consumer_model-> ThreadManagerService
 * @createDate: 2021-12-20 17:43
 * @author: ZhangSheng
 * @version: 1.0
 * @description:  线程管理服务层
 */

@Slf4j
public class ThreadManagerService {


    private LinkedBlockingDeque<Message> blockingQueue  = new LinkedBlockingDeque<Message>();

    private int workerCount = 1;

    public synchronized void add(Message messsage) throws InterruptedException {
        blockingQueue.put(messsage);
    }

    public synchronized void addLast(Message messsage){
       blockingQueue.addLast(messsage);
    }


    public void init(){
        for (int i = 0; i < workerCount; i++) {
            Thread worker  = new Thread(new Worker(blockingQueue));
            worker.start();
        }
    }


    public static class Worker implements Runnable{

        LinkedBlockingDeque<Message> blockingQueue;

        public Worker(LinkedBlockingDeque<Message> blockingQueue){
            this.blockingQueue = blockingQueue;
        }

        public Worker(){

        }

        @Override
        public void run() {
            while(true){
                try {
                    Message poll = blockingQueue.poll(2000, TimeUnit.SECONDS);
                    if (!ObjectUtils.isEmpty(poll)){
                        log.info("工作者 {}: 开始消费 {} 消息.",Thread.currentThread().getName(),poll.getValue());
                    }
                } catch (Exception e) {
                    log.info("获取队列数据异常：{}", e.getMessage());
                    continue;
                }
            }
        }
    }

}
